package org.study.flink_cdc.boot.config;

import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.springframework.stereotype.Component;

/**
 * 作用描述
 *
 * @author doveylovey
 * @version v1.0.0
 * @email 1135782208@qq.com
 * @date 2024年11月05日
 */
@Slf4j
@Component
public class FlinkCdcCommandLineRunner   {
    @PostConstruct
    public void afterPropertiesSet() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 使用 Flink-CDC 连接 MySQL
        String name = "t_flink_cdc";
        tableEnv.executeSql("CREATE TABLE IF NOT EXISTS " + name + " (" +
                " id BIGINT," +
                " username STRING," +
                " password STRING," +
                " enabled STRING," +
                " locked STRING," +
                " weight DECIMAL(10, 2)," +
                " PRIMARY KEY (id) NOT ENFORCED " +
                ") WITH (" +
                " 'connector' = 'mysql-cdc'," +
                " 'hostname' = '127.0.0.1'," +
                " 'port' = '3306'," +
                " 'username' = 'root'," +
                " 'password' = 'root'," +
                " 'database-name' = 'test'," +
                " 'table-name' = 't_user'," +
                " 'scan.incremental.snapshot.chunk.key-column' = 'id' " +
                ")");
        // 查询并打印结果
        // DataStream<String> dataStream = tableEnv.sqlQuery("SELECT * FROM " + name).execute().print();
        TableResult tableResult = tableEnv.sqlQuery("SELECT * FROM " + name).execute();
        tableResult.print();
        try {
            env.execute("Flink CDC Demo");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
